package com.hexinfo.dmpro.component.comparison.handler.impl;

import cn.hutool.core.date.DateUtil;
import cn.hutool.crypto.SecureUtil;
import com.hexinfo.dmpro.component.common.util.DataSourceUtil;
import com.hexinfo.dmpro.component.comparison.entity.TableContrastDataManagementEntity;
import com.hexinfo.dmpro.component.comparison.entity.TableContrastParamEntity;
import com.hexinfo.dmpro.component.comparison.enums.TableContrastEnums;
import com.hexinfo.dmpro.component.comparison.handler.IDataContrastHandler;
import com.hexinfo.dmpro.component.comparison.handler.IDataReadHandler;
import com.hexinfo.dmpro.component.comparison.utils.GetDataUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;

import static java.lang.Thread.sleep;

@Service
@Slf4j
public class DataReadHandler implements IDataReadHandler {

    @Resource
    IDataContrastHandler dataContrastHandler;

    @Resource
    TableContrastDataManagementEntity dataManagement;

    private static Integer maxContrastRow = 50000;

    /**
     * @Method readTargetDataToMap
     * @Param tableContrastParam
     * @Return void
     * @Description 异步流式读取 比对表数据
     * @Author Huang Yanyihe
     * @Date 2023/9/8 10:43
     * @Version V1.0
     */
    @Override
    @Async("dataReadAsyncTaskExecutor")
    public void readTargetDataToMap(TableContrastParamEntity tableContrastParam) {
        log.info("[数据比对组件] 对比表[{}]数据读取开始，开始时间：{}", tableContrastParam.getTableFlag(), DateUtil.now());
        GetDataUtil dataUtil = new GetDataUtil();
        String tableFlag = tableContrastParam.getTableFlag();
        Connection connection = null;
        ResultSet targetRs = null;
        try {
            connection = DataSourceUtil.getHiveTargetConnection();
            targetRs = dataUtil.getAllTargetQueryRs(connection, tableContrastParam);
            //循环查出插入所有数据
            while (targetRs.getRow() <= maxContrastRow && targetRs.next()) {
                //阻止失败任务继续查询
                if (dataManagement.getStatusMap().get(tableFlag).getStatus().equals(TableContrastEnums.FAIL.getValue())) {
                    break;
                }
                HashMap<Long, String> tableMap = dataManagement.getContrastMap().get(tableFlag);
                if (tableMap == null) {
                    tableMap = new HashMap<>();
                    dataManagement.getContrastMap().put(tableFlag, tableMap);
                }
                StringBuilder rowData = new StringBuilder();
                //拼接当前数据
                for (int i = 1; i <= targetRs.getMetaData().getColumnCount(); i++) {
                    if (i != 1) rowData.append(",");
                    rowData.append(targetRs.getString(i));
                }
                tableMap.put((long) targetRs.getRow(), SecureUtil.md5(rowData.toString()));
            }
        } catch (SQLException e) {
            log.error("[数据比对组件] 对比表[{}]数据读取异常：{}", tableContrastParam.getTableFlag(), e.getMessage(), e);
            tableDataReadFailHandler(tableContrastParam, e.getMessage());
        } finally {
            DataSourceUtil.close(connection, targetRs);
            log.info("[数据比对组件] 对比表[{}]数据读取结束，结束时间：{}", tableContrastParam.getTableFlag(), DateUtil.now());
        }
    }

    private void tableDataReadFailHandler(TableContrastParamEntity tableContrastParam, String errorMessage) {
        dataManagement.setFailStatus(tableContrastParam.getTableFlag());
        dataManagement.setFalseMessage(tableContrastParam.getTableFlag(), errorMessage);
    }

    /**
     * @Method readSourceDataToQueue
     * @Param tableContrastParam
     * @Return void
     * @Description 异步流式读取 源表数据并开始明细比对
     * @Author Huang Yanyihe
     * @Date 2023/9/10 22:43
     * @Version V1.0
     */
    @Override
    @Async("dataReadAsyncTaskExecutor")
    public void readSourceDataToQueue(TableContrastParamEntity tableContrastParam) {
        log.info("[数据比对组件] 源表[{}]数据读取开始，开始时间：{}", tableContrastParam.getTableFlag(), DateUtil.now());
        String tableFlag = tableContrastParam.getTableFlag();
        GetDataUtil dataUtil = new GetDataUtil();
        Connection connection = null;
        ResultSet sourceRs = null;
        //当前行数标识
        try {
            connection = DataSourceUtil.getHiveSourceConnection();
            sourceRs = dataUtil.getAllSourceQueryRs(connection, tableContrastParam);
            //等待比对Map集合第一次插入成功
            while (!dataManagement.getContrastMap().containsKey(tableFlag)) {
                sleep(10);
            }
            //循环查询插入所有数据
            while (sourceRs.getRow() <= maxContrastRow && sourceRs.next()) {
                //终止已失败任务
                if (dataManagement.getStatusMap().get(tableFlag).getStatus().equals(TableContrastEnums.FAIL.getValue())) {
                    break;
                }
                //拼接当前行数据
                StringBuilder rowData = new StringBuilder();
                //拼接当前数据
                for (int i = 1; i <= sourceRs.getMetaData().getColumnCount(); i++) {
                    if (i != 1) rowData.append(",");
                    rowData.append(sourceRs.getString(i));
                }
                String md5 = SecureUtil.md5(rowData.toString());
                //放入源数据队列
                dataContrastHandler.dataContrast(tableContrastParam.getTableFlag(), (long) sourceRs.getRow(), md5);
            }
            dataManagement.getStatusMap().get(tableFlag).setDetailsContrastEndTime(DateUtil.now());
        } catch (Exception e) {
            log.error("[数据比对组件] 源表[{}]数据读取异常：{}", tableContrastParam.getTableFlag(), e.getMessage(), e);
            tableDataReadFailHandler(tableContrastParam, e.getMessage());
        } finally {
            DataSourceUtil.close(connection, sourceRs);
            log.info("[数据比对组件] 源表[{}]数据读取结束，结束时间：{}", tableContrastParam.getTableFlag(), DateUtil.now());
        }
    }


}
